package com.yonyou.rabbitmq.producer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yonyou.rabbitmq.utils.MessageUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.platform.commons.util.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Slf4j
public class ReliableRabbitTemplate extends RabbitTemplate{

    @Autowired
    private MessageProcess messageProcess;

    public void send(String exchange, String routingKey, Message message, CorrelationData correlationData,String tenantId) throws AmqpException {
        try {
            String id = UUID.randomUUID().toString();
            String messageBody = MessageUtils.getBodyStr(message);
            String messageProperties = MessageUtils.getPropertiesStr(message);

            if(StringUtils.isNotBlank(tenantId)){
                message.getMessageProperties().getHeaders().put("tenantId",tenantId);
            }

            MessageInfo messageInfo = new MessageInfo.Builder()
                    .id(id)
                    .exchange(exchange)
                    .routingKey(routingKey)
                    .messageProperties(messageProperties)
                    .messageBody(messageBody)
                    .failReason("")
                    .tenantId(tenantId)
                    .build();

            messageProcess.getPreSend().accept(messageInfo);

            ExtCorrelationData extData = new ExtCorrelationData();
            if (correlationData != null) {
                extData.setId(correlationData.getId());
            }
            extData.setMessageInfo(messageInfo);

            try {
                super.send(exchange, routingKey, message, extData);
            } catch (Exception e) {
                //极有可能是AmqpConnectException,长时间(和retryTemplate有关)连接不上
                logger.error("ReliableRabbitTemplate.send发送消息异常，异常原因：{}",e);
                messageInfo.setFailReason(e.getMessage());
                messageProcess.getMessageNack().accept(messageInfo);
            }

        } catch (Exception e) {
            logger.error("ReliableRabbitTemplate.send保存消息异常，异常原因：{}",e);
            //出现异常,极有可能是保存消息出现异常
            throw new AmqpException(e.getMessage());
        }
    }

    public void reSend(MessageInfo messageInfo) throws IOException {
        messageProcess.getPreResend().accept(messageInfo.getId());
        byte[] body = messageInfo.getMessageBody().getBytes(StandardCharsets.UTF_8);
        ObjectMapper mapper = new ObjectMapper();
        MessageProperties messageProperties = mapper.readValue(messageInfo.getMessageProperties(), MessageProperties.class);
        Message message = new Message(body, messageProperties);

        ExtCorrelationData extData = new ExtCorrelationData();
        extData.setMessageInfo(messageInfo);

        try {
            super.send(messageInfo.getExchange(), messageInfo.getRoutingKey(), message, extData);
        } catch (Exception e) {
            messageInfo.setFailReason(e.getMessage());
            messageProcess.getMessageNack().accept(messageInfo);
        }
    }

    public MessageProcess getMessageProcess() {
        return messageProcess;
    }

    public void setMessageProcess(MessageProcess messageProcess) {
        this.messageProcess = messageProcess;
    }

}
